package com.cskaoyan.kpi.output;

import com.cskaoyan.kpi.Summary;
import com.cskaoyan.kpi.util.JDBCUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @ClassName MysqlOutputFormat
 * @Description: TODO
 * @Author 远志 zhangsong@cskaoyan.onaliyun.com
 * @Date 2022/1/18 15:52
 * @Version V1.0
 **/
public class MysqlOutputFormat extends OutputFormat<Summary, Text> {

    protected static class MysqlRecordWriter extends RecordWriter<Summary, Text>{

        private Connection connection = null;

        public MysqlRecordWriter(){
            connection = JDBCUtil.getConnection();
        }

        @Override
        public void write(Summary summary, Text value) throws IOException, InterruptedException {
            PreparedStatement preparedStatement = null;
            String sql = "insert into summary values (null,?,?,?,?)";
            try {
                preparedStatement = connection.prepareStatement(sql);
                preparedStatement.setString(1, summary.getDate());
                preparedStatement.setString(2, summary.getAttribute());
                preparedStatement.setString(3, summary.getValue());
                preparedStatement.setString(4, summary.getAddition());
                preparedStatement.executeUpdate();
            } catch (SQLException e) {
                e.printStackTrace();
            }finally {
                if(preparedStatement != null){
                    try {
                        preparedStatement.close();
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }
            }

        }

        @Override
        public void close(TaskAttemptContext context) throws IOException, InterruptedException {
            if(connection != null){
                try {
                    connection.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    @Override
    public RecordWriter<Summary, Text> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
        return new MysqlRecordWriter();
    }

    @Override
    public void checkOutputSpecs(JobContext context) throws IOException, InterruptedException {

    }

    private FileOutputCommitter committer = null;
    public static Path getOutputPath(JobContext job) {
        String name = job.getConfiguration().get("mapred.output.dir");
        return name == null?null:new Path(name);
    }
    @Override
    public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException {
        if (committer == null) {
            Path output = getOutputPath(context);
            committer = new FileOutputCommitter(output, context);
        }
        return committer;
    }
}
